package net.bwie.realtime.jtp.ods;

import bean.CouponTransaction;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.time.Instant;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class TransactionDataGenerator {

    // 交易事件POJO类


    public static void main(String[] args) throws Exception {
        // 1. 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  // 控制数据生成速率

        // 2. 创建模拟数据源
        DataStream<CouponTransaction> transactionStream = env.addSource(
                new SourceFunction<CouponTransaction>() {
                    private volatile boolean isRunning = true;
                    private final Random random = new Random();

                    // 预定义模拟数据选项
                    private final List<String> provinces = Arrays.asList(
                            "北京", "上海", "广东", "江苏", "浙江", "四川", "湖北");
                    private final List<String> categories = Arrays.asList(
                            "电子产品", "家用电器", "服装鞋帽", "美妆个护", "食品生鲜");
                    private final List<String> cities = Arrays.asList(
                            "北京市", "上海市", "广州市", "深圳市", "杭州市", "成都市", "武汉市");

                    @Override
                    public void run(SourceContext<CouponTransaction> ctx) throws Exception {
                        while (isRunning) {
                            CouponTransaction event = generateRandomTransaction();
                            ctx.collect(event);
                            TimeUnit.MILLISECONDS.sleep(100);  // 控制生成速率
                        }
                    }

                    private CouponTransaction generateRandomTransaction() {
                        CouponTransaction event = new CouponTransaction();

                        // 生成随机数据
                        event.setTransactionId(UUID.randomUUID().toString());
                        event.setUserId((long) random.nextInt(1000000));
                        event.setItemId((long) random.nextInt(10000));
                        event.setItemName("商品-" + random.nextInt(100));
                        event.setItemCategory(categories.get(random.nextInt(categories.size())));

                        // 价格相关字段
                        double originalPrice = 100 + random.nextDouble() * 900; // 100-1000
                        event.setOriginalPrice(Math.round(originalPrice * 100.0) / 100.0);

                        double coupon = random.nextDouble() * 50; // 0-50优惠券
                        event.setCouponAmount(Math.round(coupon * 100.0) / 100.0);

                        event.setPaymentAmount(Math.round((originalPrice - coupon) * 100.0) / 100.0);

                        // 地理位置
                        int provinceIndex = random.nextInt(provinces.size());
                        event.setProvince(provinces.get(provinceIndex));
                        event.setCity(cities.get(provinceIndex));  // 简单对应关系

                        // 时间戳
                        event.setTimestamp(Instant.now().toEpochMilli());

                        // 优惠券ID
                        event.setPromotionId("PROMO-" + (random.nextBoolean() ? "DISCOUNT" : "CASHBACK"));

                        return event;
                    }

                    @Override
                    public void cancel() {
                        isRunning = false;
                    }
                }
        );

        // 3. 序列化为JSON字符串
        DataStream<String> jsonStream = transactionStream
                .map(event -> {
                    // 手动构建JSON（推荐使用Jackson等库）
                    return String.format(
                            "{\"transactionId\":\"%s\",\"userId\":%d,\"itemId\":%d," +
                                    "\"itemName\":\"%s\",\"itemCategory\":\"%s\",\"originalPrice\":%.2f," +
                                    "\"couponAmount\":%.2f,\"paymentAmount\":%.2f,\"province\":\"%s\"," +
                                    "\"city\":\"%s\",\"timestamp\":%d,\"promotionId\":\"%s\"}",
                            event.getTransactionId(), event.getUserId(), event.getItemId(),
                            event.getItemName(), event.getItemCategory(), event.getOriginalPrice(),
                            event.getCouponAmount(), event.getPaymentAmount(), event.getProvince(),
                            event.getCity(), event.getTimestamp(), event.getPromotionId()
                    );
                });

        // 4. 配置Kafka Producer

        KafkaUtil.producerKafka(jsonStream,"jd-topic");

        // 6. 执行作业
        env.execute("JD 618 Transaction Data Generator");
    }
}